package org.project.d.network;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.project.d.network.msg.CommandCode;
import org.project.d.network.msg.Transmission;
import org.project.d.network.msg.enums.SerializationType;
import org.project.d.network.msg.serializer.Serialized;
import org.project.d.network.msg.serializer.SerializationManager;
import org.project.d.network.protocol.Packet;
import org.project.d.network.protocol.UDPPacket;

import java.net.InetSocketAddress;
import java.net.SocketAddress;

/**
 * @author lin
 * @version V1.0
 * @description
 * @date 2020/6/18 4:04 下午
 */
@ToString
@Slf4j
public class NettyConnection implements Connection, ChannelFutureListener {
    //    private static final Logger LOGGER = LoggerFactory.getLogger(NettyConnection.class);
    private Channel channel;
    private volatile byte status = STATUS_NEW;
    private long lastReadTime;
    private long lastWriteTime;
    public int heartbeat = Integer.parseInt(System.getProperty("project.d.heartbeat", "10000"));

    private SocketAddress inetSocketAddress;

    private Byte channelType;
    private Byte serializationType;

    @Override
    public void init(Channel channel, SocketAddress inetSocketAddress, Byte channelType, Byte serializationType) {
        this.channel = channel;
        this.lastReadTime = System.currentTimeMillis();
        this.status = STATUS_CONNECTED;
        this.inetSocketAddress = inetSocketAddress;
        this.channelType=channelType;
        this.serializationType=serializationType;
    }

    @Override
    public ChannelFuture send(Transmission transmission) {
        return send(transmission, null);
    }

    @Override
    public ChannelFuture send(TransmissionBuild transmission) {
        return send(transmission,null);
    }


    @Override
    public ChannelFuture send(Transmission transmission, ChannelFutureListener listener) {
        if (channel.isActive()) {
            CommandCode commandCode = transmission.getCommandCode();
            Byte serializationType = getSerializationType(transmission);
            Serialized serializer = getSerializa(transmission, serializationType);
            byte[] bytes = serializer.serialized(transmission.getMessage());
            Packet packet = new Packet(Config.rpcNum, commandCode.value(), Config.flags, serializationType, Config.lrc, bytes.length, bytes);
            if (channelType == Connection.CHANNEL_TYPE_UDP) {
                if (inetSocketAddress instanceof InetSocketAddress){
                    packet = new UDPPacket(Config.rpcNum, commandCode.value(), Config.flags, serializationType, Config.lrc, bytes.length, bytes, ((InetSocketAddress) inetSocketAddress));
                }else {
                    throw new RuntimeException("if UDPPacket please use InetSocketAddress");
                }
            }
            ChannelFuture channelFuture = channel.writeAndFlush(packet);
            ChannelFuture future = channelFuture.addListener(this);
            if (listener != null) {
                future.addListener(listener);
            }

            if (channel.isWritable()) {
                return future;
            }

            //阻塞调用线程还是抛异常？
            //return channel.newPromise().setFailure(new RuntimeException("send data too busy"));
            if (!future.channel().eventLoop().inEventLoop()) {
                future.awaitUninterruptibly(100);
            }
            return future;
        } else {
            log.warn("this channel is not active :{},now close this channel",channel);
            /*if (listener != null) {
                channel.newPromise()
                        .addListener(listener)
                        .setFailure(new RuntimeException("connection is disconnected"));
            }*/
            return this.close();
        }
    }

    private Byte getSerializationType(Transmission transmission) {
        Byte serializationType = this.serializationType;
        if (serializationType ==null){
            serializationType = Config.serializationType;
        }
        if (transmission.getSerializationType()!=null){
            serializationType =transmission.getSerializationType();
        }
        if (transmission.getSerialized()!=null){
            serializationType=SerializationType.UNKNOWN.getType();
        }
        return serializationType;
    }

    private Serialized getSerializa(Transmission transmission, Byte serializationType) {
        Serialized serializer;
        if (transmission.getSerialized()!=null){
            serializer=transmission.getSerialized();
        }else {
            serializer= SerializationManager.getSerializer(serializationType);
        }
        return serializer;
    }

    @Override
    public ChannelFuture send(TransmissionBuild transmission, ChannelFutureListener listener) {
        Transmission.TransmissionBuilder builder = Transmission.builder();
        return send(transmission.bulid(builder).build(),listener);
    }


    @Override
    public String getId() {
        return channel.id().asShortText();
    }

    @Override
    public ChannelFuture close() {
        if (status == STATUS_DISCONNECTED) return null;
        this.status = STATUS_DISCONNECTED;
        return this.channel.close();
    }

    @Override
    public boolean isConnected() {
        return status == STATUS_CONNECTED;
    }

    @Override
    public boolean isReadTimeout() {
        return System.currentTimeMillis() - lastReadTime > heartbeat + 1000;
    }

    @Override
    public boolean isWriteTimeout() {
        return System.currentTimeMillis() - lastWriteTime > heartbeat - 1000;
    }

    @Override
    public void updateLastReadTime() {
        lastReadTime = System.currentTimeMillis();
    }

    @Override
    public void updateLastWriteTime() {
        lastWriteTime = System.currentTimeMillis();
    }

    @Override
    public Channel getChannel() {
        return channel;
    }

    @Override
    public SocketAddress getAddress() {
        return inetSocketAddress;
    }

    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
        if (future.isSuccess()) {
            lastWriteTime = System.currentTimeMillis();
        } else {
//            LOGGER.error("connection send msg error", future.cause());
            log.error("connection send msg error={"+future.cause().getMessage()+"}, conn={"+this+"}", future.cause());
        }
    }
}
